package com.walkud.rxjava.filtering;

import com.walkud.rxjava.utils.PrintLogSubscriber;
import com.walkud.rxjava.utils.RxSubscribe;
import com.walkud.rxjava.utils.SleepUtil;
import io.reactivex.MaybeObserver;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Predicate;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 过滤操作符示例
 * Debounce — only emit an item from an Observable if a particular timespan has passed without it emitting another item
 * Distinct — suppress duplicate items emitted by an Observable
 * ElementAt — emit only item n emitted by an Observable
 * Filter — emit only those items from an Observable that pass a predicate test
 * First — emit only the first item, or the first item that meets a condition, from an Observable
 * IgnoreElements — do not emit any items from an Observable but mirror its termination notification
 * Last — emit only the last item emitted by an Observable
 * Sample — emit the most recent item emitted by an Observable within periodic time intervals
 * Skip — suppress the first n items emitted by an Observable
 * SkipLast — suppress the last n items emitted by an Observable
 * Take — emit only the first n items emitted by an Observable
 * TakeLast — emit only the last n items emitted by an Observable
 */
public class FilteringSimple {

    public static void main(String[] args) {
//        debounce();
//        distinct();
//        elementat();
//        filter();
//        first();
//        ignoreElements();
//        last();
//        sample();
//        skip();
//        skipLast();
//        take();
        takeLast();
    }

    /**
     * 只发射后面的N项数据（与SkipLast相反）
     */
    private static void takeLast() {
        Observable.just(1, 2, 3, 4, 5).takeLast(2).subscribe(PrintLogSubscriber.create());
    }

    /**
     * 只发射前面的N项数据（与Skip相反）
     */
    private static void take() {
        Observable.just(1, 2, 3, 4, 5).take(2).subscribe(PrintLogSubscriber.create());
    }

    /**
     * 跳过Observable发射的后N项数据
     */
    private static void skipLast() {
        //跳过4，5
        Observable.just(1, 2, 3, 4, 5).skipLast(2).subscribe(PrintLogSubscriber.create());
    }

    /**
     * 跳过Observable发射的前N项数据
     */
    private static void skip() {
        //跳过1，2
        Observable.just(1, 2, 3, 4, 5).skip(2).subscribe(PrintLogSubscriber.create());
    }

    /**
     * 定期发射Observable最近发射的数据项
     */
    private static void sample() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                for (int i = 0; i < 10; i++) {
                    Thread.sleep(400);
                    emitter.onNext(i);
                }

                emitter.onComplete();
            }
        }).sample(1000, TimeUnit.MILLISECONDS)
                .subscribe(PrintLogSubscriber.create());
    }

    /**
     * 获取最后一个元素（类似于List.get(list.size()-1),但有效防止下标越界，因为有默认值）
     */
    private static void last() {
        Observable.just(1, 2, 3, 4, 5).last(0).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                System.out.print(integer);
            }
        });
//        Observable.empty().last(0).subscribe(new Consumer<Object>() {
//            @Override
//            public void accept(Object o) throws Exception {
//                System.out.print(o);
//            }
//        });
    }

    /**
     * 忽略元素发送（觉得没什么用）
     */
    private static void ignoreElements() {
        //模拟延迟1秒执行，忽略元素发送，直接运行
        Observable.timer(1, TimeUnit.SECONDS).ignoreElements().subscribe(new Action() {
            @Override
            public void run() {
                System.out.println("run");
            }
        });

        //防止线程提前结束
        SleepUtil.sleep();
    }

    /**
     * 获取第一个元素(类似于List.get(0),有效防止小标越界)
     */
    private static void first() {
        Observable.just(1, 2, 3, 4, 5).first(0).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                System.out.println(integer);
            }
        });

//        Observable.empty().first(0).subscribe(new Consumer<Object>() {
//            @Override
//            public void accept(Object o) throws Exception {
//                System.out.print(o);
//            }
//        });
    }

    /**
     * 过滤操作符
     */
    private static void filter() {
        Observable.just(1, 2, 3, 4, 5, 6).filter(new Predicate<Integer>() {
            @Override
            public boolean test(Integer integer) {
                return integer % 2 == 0;//过滤掉奇数，保留偶数
            }
        }).subscribe(PrintLogSubscriber.create());
    }

    /**
     * 获取集合元素对应下标元素，找到对应元素调用onSuccess()，找不到对应元素，则调用onComplete(),可以有效防止下标越界，但下标必须大于等于0
     */
    private static void elementat() {
        List<Integer> list = Arrays.asList(1, 2, 3, 4, 5);
        Observable.fromIterable(list).elementAt(10).subscribe(new MaybeObserver<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("onSubscribe()");
            }

            @Override
            public void onSuccess(Integer t) {
                System.out.println("onSuccess():" + t);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("onError():" + e.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete()");
            }
        });
    }

    /**
     * 去重操作符
     */
    private static void distinct() {
        Observable.just(1, 3, 4, 1, 2, 3).distinct().subscribe(PrintLogSubscriber.create());
    }

    /**
     * 仅在过了一段指定的时间还没发射数据时才发射一个数据
     * 使用场景：及时输入搜索场景
     */
    private static void debounce() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) {
                //模拟用户连续输入
                for (int i = 0; i < 10; i++) {
                    emitter.onNext(i);
                }
                //模拟用户停顿1秒
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                //模拟用户继续连续输入
                for (int i = 10; i < 15; i++) {
                    emitter.onNext(i);
                }

                //用户输入完成
                emitter.onComplete();
            }
        }).debounce(500, TimeUnit.MILLISECONDS)//使用debounce过滤500毫秒内抖动
                .subscribe(PrintLogSubscriber.create());
    }
}
